package flink;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class sqlUtils {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env,settings);

        //读取文件
        String sql = "CREATE TABLE MyUserTable (\n" +
                "  source STRING,\n" +
                "  columnName STRING,\n" +
                "  columnType STRING\n" +
                ") WITH (\n" +
                "  'connector' = 'filesystem',\n" +
                "  'path' = 'C:\\Users\\Administrator\\Desktop\\data.txt'," +
                "  'format' = 'csv')";
        tableEnv.executeSql(sql);
        Table table = tableEnv.sqlQuery("select * from MyUserTable");
        table.printSchema();
        tableEnv.toRetractStream(table, Row.class).print();
        env.execute();
    }
}
